package com.wzy.product;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @Author: wzy
 * @Date: 2022/03/27/15:56
 * @Description:
 */
public class MyProducer {
    private final static String TOPIC_NAME = "my-replicated-topics";

    public static void main(String[] args) throws ExecutionException,
            InterruptedException {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "192.168.188.136:9092,192.168.188.136:9093,192.168.188.136:9094");

        //把发送的key从字符串序列化为字节数组
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class.getName());
        //把发送消息value从字符串序列化为字节数组
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class.getName());
        Producer<String, String> producer = new KafkaProducer<String,
                String>(props);
        ProducerRecord<String, String> producerRecord = new
                ProducerRecord<String, String>(TOPIC_NAME,0
                , "wzy002", "hello kafka");
        //等待消息发送成功的同步阻塞⽅法
        RecordMetadata metadata = producer.send(producerRecord).get();
        //=====阻塞=======
        System.out.println("同步⽅式发送消息结果：" + "topic-" +
                metadata.topic() + "|partition-"
                + metadata.partition() + "|offset-" +
                metadata.offset());

    }
}
